package com.atguigu.sparkstreaming.getoffsets

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/8/22
 *
 *
 *
 *      探究偏移量获取的位置(在Driver端 还是在Executor端获取)？
 *
 *          规律：   foreachRDD 和 transform中编写的代码，只有在RDD.算子( xxx ),xxx才是在Executor端运行，其余部分都是在Driver端运行。
 *
 */
object GetOffsetsDemo3 {

  def main(args: Array[String]): Unit = {


    val streamingContext = new StreamingContext("local[*]", "wordcount", Seconds(5))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "sz220409test",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    val topics = Array("topicD")


    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    ds.foreachRDD(rdd => {

      // asInstanceOf是scala中为所有的对象都提供的一个强转运算符，并不是RDD中自己编写的算子，不要混淆
      val ranges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      // 1.streaming-job-executor-0   driver
      for (offsetRange <- ranges) {
        println(Thread.currentThread().getName + "当前消费了:" + offsetRange.topic + "  主题的"
          + offsetRange.partition + "  分区，从"
          + offsetRange.fromOffset + " 消费到了 " + offsetRange.untilOffset)
      }

      //进行输出
      //2. Executor task launch worker for task 24   executor
      rdd.foreach(record => println(Thread.currentThread().getName + record.value()));

    })


    streamingContext.start()

    streamingContext.awaitTermination()

  }

}
